#pragma once
#include <iostream>
#include <string>
#include "../mqcommon/Helper.hpp"
#include "../mqcommon/Log.hpp"
#include "muduo/net/TcpServer.h"
#include "muduo/net/EventLoop.h"
#include "../third/protobuf/codec.h"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/proto.pb.h"
#include "Consumer.hpp"
#include "Host.hpp"
#include "Route.hpp"
#include "../mqcommon/ThreadPool.hpp"

using namespace std;
namespace mq
{
    using openChannelRequestPtr = shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = shared_ptr<deleteQueueRequest>;
    using BindRequestPtr = shared_ptr<BindRequest>;
    using unBindRequestPtr = shared_ptr<unBindRequest>;
    using basicPublishRequestPtr = shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = shared_ptr<basicCancelRequest>;
    using basicConsumeResponsePtr = shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = shared_ptr<basicCommonResponse>;
    using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;
    class Channel
    {
    public:
        using CallBackType = function<void(const string &, const BasicProperties *bp, const string &)>;
        using ChannelPtr = shared_ptr<Channel>;
        Channel(const string &id, const VirtualHost::VirtualHostPtr &host,
                const ConsumerManager::ConsumerManagerPtr &cmp, const ProtobufCodecPtr &codec,
                const muduo::net::TcpConnectionPtr &conn, const ThreadPool::ThreadPoolPtr &pool)
            : _cid(id), _host(host), _cmp(cmp), _codec(codec), _conn(conn), _pool(pool) {}

        ~Channel()
        {
            if (_consumer.get() != nullptr)
                _cmp->remove(_consumer->_tag, _consumer->_qname);
        }
        // 声明交换机
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _host->declareExchange(req->ename(), req->type(), req->durable(), req->auto_delete(), req->args());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 删除交换机
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->ename());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 声明队列
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _host->declareQueue(req->qname(), req->durable(), req->exclusive(), req->auto_delete(), req->args());
            if (ret == false)
                return basicResponse(ret, req->rid(), req->cid());
            _cmp->initQueueConsumer(req->qname());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 删除队列
        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            _host->deleteQueue(req->qname());
            _cmp->destroyQueueConsumer(req->qname());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 添加绑定信息
        void queueBind(const BindRequestPtr &req)
        {
            bool ret = _host->Bind(req->ename(), req->qname(), req->binding_key());
            return basicResponse(ret, req->rid(), req->cid());
        }
        // 删除绑定信息
        void queueUnBind(const unBindRequestPtr &req)
        {
            _host->unBind(req->ename(), req->qname());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 发布消息
        void basicPublish(const basicPublishRequestPtr &req)
        {
            Exchange::ExchangePtr ep = _host->selectExchange(req->ename());
            if (ep.get() == nullptr)
            {
                DLOG("目标交换机不存在:%s", req->ename().c_str());
                return basicResponse(false, req->rid(), req->cid());
            }
            BindingManager::MsgQueueBindingMap mqbm = _host->exchangeBindings(req->ename());
            bool ret;
            BasicProperties *properties = nullptr;
            string routing_key;
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            for (auto &it : mqbm)
            {
                // DLOG("获取到%s队列的绑定关系%s", it.first.c_str(), it.second->_binding_key.c_str());
                // DLOG("routing_key: %s binding_key: %s", routing_key.c_str(), it.second->_binding_key.c_str());
                // DLOG("交换机模式: %d", ep->_type);
                // DLOG("条件判断: %d %d %d", Router::isLegalBindingKey(it.second->_binding_key), Router::isLegalRoutingKey(routing_key), Router::route(ep->_type, routing_key, it.second->_binding_key));
                if (Router::isLegalBindingKey(it.second->_binding_key) &&
                    Router::isLegalRoutingKey(routing_key) &&
                    Router::route(ep->_type, routing_key, it.second->_binding_key))
                {
                    ret = _host->basicPublish(it.first, req->mutable_properties(), req->body());
                    if (ret == false)
                    {
                        return basicResponse(ret, req->rid(), req->cid());
                    }
                    auto task = bind(&Channel::consume, this, it.first);
                    _pool->push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }
        // 确认消息
        void basicAck(const basicAckRequestPtr &req)
        {
            _host->basicAck(req->qname(), req->msg_id());
            return basicResponse(true, req->rid(), req->cid());
        }
        // 订阅消息
        void basicConsume(const basicConsumeRequestPtr &req)
        {
            bool ret = _host->existsQueue(req->qname());
            if (ret == false)
            {
                DLOG("目标队列不存在: %s", req->qname().c_str());
                return basicResponse(false, req->rid(), req->cid());
            }
            auto func = bind(&Channel::callback, this, placeholders::_1, placeholders::_2, placeholders::_3);
            _consumer = _cmp->create(req->tag(), req->qname(), req->auto_ack(), func);
            return basicResponse(true, req->rid(), req->cid());
        }
        // 取消订阅消息
        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->remove(req->tag(), req->qname());
            return basicResponse(true, req->rid(), req->cid());
        }

    private:
        void callback(const string &tag, const BasicProperties *bp, const string &body)
        {
            basicConsumeResponse resp;
            resp.set_tag(tag);
            resp.set_cid(_cid);
            resp.set_body(body);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            }
            _codec->send(_conn, resp);
        }
        void consume(const string &qname)
        {
            Consumer::ConsumerPtr cs = _cmp->choose(qname);
            if (cs.get() == nullptr)
            {
                DLOG("执行消费任务失败，%s 队列没有消费者", qname.c_str());
                return;
            }
            MessageManager::MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                DLOG("执行消费任务失败，%s 队列没有消息", qname.c_str());
                return;
            }
            cs->_callback(cs->_tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            if (cs->_auto_ack == true)
            {
                _host->basicAck(qname, mp->payload().properties().id());
            }
        }
        void basicResponse(bool ok, const string &rid, const string &cid)
        {
            basicCommonResponse resp;
            resp.set_ok(ok);
            resp.set_rid(rid);
            resp.set_cid(cid);
            _codec->send(_conn, resp);
        }

    private:
        string _cid;
        Consumer::ConsumerPtr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ConsumerManager::ConsumerManagerPtr _cmp;
        VirtualHost::VirtualHostPtr _host;
        ThreadPool::ThreadPoolPtr _pool;
    };
    class ChannelManager
    {
    public:
        using ChannelManagerPtr = shared_ptr<ChannelManager>;
        ChannelManager()
        {
        }
        // 新建信道
        bool openChannel(const string &id, const VirtualHost::VirtualHostPtr &host, const ConsumerManager::ConsumerManagerPtr &cmp,
                         const ProtobufCodecPtr &codec, const muduo::net::TcpConnectionPtr &conn, const ThreadPool::ThreadPoolPtr &pool)
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it != _channels.end())
            {
                DLOG("信道已存在: %s", id.c_str());
                return false;
            }
            Channel::ChannelPtr channel = make_shared<Channel>(id, host, cmp, codec, conn, pool);
            _channels.insert(make_pair(id, channel));
            // DLOG("已建立信道: %s", id.c_str());
            return true;
        }
        // 删除信道
        void closeChannel(const string &id)
        {
            unique_lock<mutex> lock(_mutex);
            _channels.erase(id);
        }
        // 获取信道操作句柄
        Channel::ChannelPtr getChannel(const string &id)
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                DLOG("信道不存在: %s", id.c_str());
                return Channel::ChannelPtr();
            }
            return it->second;
        }

    private:
        mutex _mutex;
        unordered_map<string, Channel::ChannelPtr> _channels;
    };
}